// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: donggu

#include <fmt/format.h>

#include "coin2/exchange/base/feed/live/system.h"

#include "coin2/exchange/base/feed/ws/subscriber_manager.h"
#include "coin2/exchange/di/constant.h"
#include "coin2/exchange/factory/symbology/product.h"
#include "coin2/flow/symbol_group.h"

namespace coin2::exchange::base::feed {

LiveFeedSubsystem::LiveFeedSubsystem(
    const std::string& system_name,
    const MarketExchangeApi& mea,
    const coin2::exchange::di::ConstantInjection& ci,
    const coin2::exchange::feed::FeedSystemConfig& config,
    coin2::exchange::base::executor::LiveAsyncExecutor* live_executor,
    IFeedSubscriber* callback)
    : BaseLiveFeedSubsystem(system_name, mea, live_executor),
      num_workers_(config.websocket().num_workers()),
      feed_arbitration_(config.websocket().arbitration()),
      ci_(ci) {
  processor_ = factory::feed::NewRawProcessor(
      ci, system_name, mea, config.raw_parser(), config.recipe());
  if (feed_arbitration_) {
    LOG(INFO) << "[LiveFeedSubsystem] Feed arbitration is enabled";
    const auto& c = config.websocket();
    CHECK_GE(c.num_workers(), 1) << c.DebugString();
    CHECK(!c.has_worker_id()) << c.DebugString();
    subscriber_ = std::make_shared<SubscriberManager>(
        ci.GetFeedSubscriptionSpec(mea),
        config.websocket(),
        coin2::flow::TopicType::LONG_TOPIC,
        live_executor,
        config.raw_parser().enable_telemetry());
  } else {
    LOG(INFO) << "[LiveFeedSubsystem] Feed arbitration is disabled";
    subscriber_ = std::make_shared<SubscriberManager>(
        ci.GetFeedSubscriptionSpec(mea),
        config.websocket(),
        coin2::flow::TopicType::SHORT_TOPIC,
        live_executor,
        config.raw_parser().enable_telemetry());
  }

  auto temp_products = ProductRequestsFromConfig(
      mea_.String(),
      config.products(),
      GetCurrentTimestamp(),
      false);  // die_on_failure
  if (config.products().has_norm_format()
      && config.products().norm_format().find("*") != std::string::npos) {
    products_ = FilterProductWithInSymbolGroups(mea_, temp_products);
  } else {
    products_ = std::move(temp_products);
  }

  for (auto& prod : products_) {
    auto symbol = prod->feed_native_symbol();
    char group = coin2::flow::SymbolGroupConfig::GetInstance().GetGroupForRelativeNorm(
        mea_,
        prod->relative_norm());
    subscriber_->RequestNativeSymbol(symbol, group);
    processor_->AddProduct(*prod, callback);
  }
}

void LiveFeedSubsystem::StartAsync() {
  if (feed_arbitration_) {
    topic_register_.RegisterLiveLongTopicArb(
        mea_, processor_.get(), live_executor_->distributor(),
        ci_.GetArbitrationConfigs(), num_workers_);
  } else {
    if (mea_.exchange == coin::proto::ExchangeType::Bitflyer) {
      // Special handling to Spot.Bitflyer
      MarketExchangeApi mea_tmp{
          coin::proto::MarketType::Futures,
          coin::proto::ExchangeType::Bitflyer,
          "v1"};
      topic_register_.RegisterLiveShortTopic(
          mea_tmp, processor_.get(), live_executor_->distributor());
    } else if (mea_.exchange == coin::proto::ExchangeType::Prex) {
      // Prex case always use long topic name.
      topic_register_.RegisterLiveLongTopic(
          mea_, processor_.get(), live_executor_->distributor());
    } else {
      topic_register_.RegisterLiveShortTopic(
          mea_, processor_.get(), live_executor_->distributor());
    }
  }
  subscriber_->RegisterPubTopic();
  subscriber_->StartAsync();
  health_.status = HealthStatus::HEALTH_STARTING;
  health_.since = GetCurrentTimestamp();
}

void LiveFeedSubsystem::StopAsync() { subscriber_->StopAsync(); }

std::vector<IProduct*> LiveFeedSubsystem::GetProducts() const {
  std::vector<IProduct*> out;
  for (const auto& p : products_) {
    out.push_back(p.get());
  }
  return out;
}

void LiveFeedSubsystem::GetHealthReport(int indent, std::string* report) const {
  // [HealthReport]
  //     [Feed]
  //         - Huobi (OK)
  //         - Okex (STARTING, uptime=23h)
  //               a1 (..)
  HealthReport health{};
  subscriber_->GetHealthReport(indent + 1, &health);

  report->append(std::string(indent * 4, ' '));
  std::string info = fmt::format(
      "- {system} ({health}, uptime='{uptime}', alive_connection={connection_num}, symbols={symbol_num})\n",
      fmt::arg("system", system_name()),
      fmt::arg("health", HealthStatusName(this->health_.status)),
      fmt::arg("uptime", DebugStringFromDuration(GetCurrentTimestamp() - health_.since)),
      fmt::arg("connection_num", health.connection_num),
      fmt::arg("symbol_num", health.symbol_num));
  report->append(info).append(health.report);
}

bool LiveFeedSubsystem::BanProduct(const IProduct* product) {
  std::vector<const IProduct*> products{product};
  return BanProduct(products);
}

bool LiveFeedSubsystem::BanProduct(const std::vector<const IProduct*>& products) {
  std::vector<const IProduct*> checked_products;
  for (auto& product : products) {
    if (product->market() == mea_.market &&
        product->exchange() == mea_.exchange) {
      checked_products.push_back(product);
    }
  }

  if (checked_products.empty()) return false;

  processor_->RemoveProduct(checked_products);
  subscriber_->BanProduct(checked_products);

  return true;
}

bool LiveFeedSubsystem::BanProduct(const std::vector<IProduct*>& products) {
  std::vector<const IProduct*> const_products;
  const_products.insert(const_products.end(), products.begin(), products.end());
  return BanProduct(const_products);
}

}  // namespace coin2::exchange::base::feed
